package org.softee.functional;

import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

import org.softee.functional.functions.Mapper;
import org.softee.functional.functions.Predicate;
import org.softee.functional.functions.ReduceFunction;
import org.softee.functional.functions.SequentialOnly;

/**
 * Parallel strategy: Atomize the tasks into individual payloads and perform computation on each using executor.
 *
 * @author morten.hattesen@gmail.com
 *
 */
public class ParallelSegmented<F> extends AbstractParallelFunctional<F> {
    /**
     * The minimum number of elements before attempting to split when performing a reduce function
     */
    private static final int REDUCE_PARALLEL_SIZE_LIMIT = 4;
    /**
     * The minimum number of elements in each split data collection
     */
    private static final int REDUCE_SEGMENT_SIZE_LIMIT = REDUCE_PARALLEL_SIZE_LIMIT / 2;

    public ParallelSegmented(Collection<F> data) {
        super(data);
    }

    public ParallelSegmented(Collection<F> data, ExecutionContext executionContext) {
        super(data, executionContext);
    }

    @Override
    public ParallelFunctional<F> filter(final Predicate<? super F> predicate) {
        if (data.size() <= 1 || predicate instanceof SequentialOnly) {
            return withData(doFilter(predicate, data));
        }

        // split data
        Collection<Collection<F>> segments = executionContext.split(data, 1);

        // optimize - no point in doing async computation on one segment
        if (segments.size() == 1) {
            return withData(doFilter(predicate, data));
        }

        // execute in parallel
        Collection<Future<Collection<F>>> futureSegmentResults = new ArrayList<Future<Collection<F>>>(segments.size());
        for (final Collection<F> segment : segments) {
            Callable<Collection<F>> callable = new Callable<Collection<F>>() {
                @Override
                public Collection<F> call() throws Exception {
                    return doFilter(predicate, segment);
                }
            };
            FutureTask<Collection<F>> futureTask = new FutureTask<Collection<F>>(callable);
            executionContext.getExecutor().execute(futureTask);
            futureSegmentResults.add(futureTask);
        }

        // merge future results
        Collection<F> results = new ArrayList<F>(data.size());
        for (Future<Collection<F>> future : futureSegmentResults) {
            try {
                Collection<F> segmentResults = future.get();
                results.addAll(segmentResults);
            } catch (InterruptedException e) {
                // propagate
                Thread.currentThread().interrupt();
                abort(e, futureSegmentResults);
            } catch (ExecutionException e) {
                abort(e.getCause(), futureSegmentResults);
            }
        }

        return withData(results);
    }

    @Override
    public <T> ParallelFunctional<T> map(final Mapper<? super F, ? extends T> mapper) {
        if (data.size() <= 1  || mapper instanceof SequentialOnly) {
            return withData(doMap(mapper, data));
        }

        // split data
        Collection<Collection<F>> segments = executionContext.split(data, 1);

        // optimize - no point in doing async computation on one segment
        if (segments.size() == 1) {
            return withData(doMap(mapper, data));
        }

        // execute in parallel
        Collection<Future<Collection<T>>> futureSegmentResults = new ArrayList<Future<Collection<T>>>(segments.size());
        for (final Collection<F> segment : segments) {
            Callable<Collection<T>> callable = new Callable<Collection<T>>() {
                @Override
                public Collection<T> call() throws Exception {
                    return doMap(mapper, segment);
                }
            };
            FutureTask<Collection<T>> futureTask = new FutureTask<Collection<T>>(callable);
            executionContext.getExecutor().execute(futureTask);
            futureSegmentResults.add(futureTask);
        }

        // merge future results
        Collection<T> results = new ArrayList<T>(data.size());
        for (Future<Collection<T>> future : futureSegmentResults) {
            try {
                Collection<T> segmentResults = future.get();
                results.addAll(segmentResults);
            } catch (InterruptedException e) {
                // propagate
                Thread.currentThread().interrupt();
                abort(e, futureSegmentResults);
            } catch (ExecutionException e) {
                abort(e.getCause(), futureSegmentResults);
            }
        }

        return withData(results);
    }

    @Override
    public F reduce(final ReduceFunction<F> reduceFunction) {
        if (data.size() < REDUCE_PARALLEL_SIZE_LIMIT || reduceFunction instanceof SequentialOnly) {
            // perform the reduction sequentially
            return doReduce(reduceFunction, data);
        }

        // split data
        Collection<Collection<F>> segments = executionContext.split(data, REDUCE_SEGMENT_SIZE_LIMIT);

        // optimize - no point in doing async computation on one segment
        if (segments.size() == 1) {
            return doReduce(reduceFunction, data);
        }

        // execute in parallel
        Collection<Future<F>> futureSegmentResults = new ArrayList<Future<F>>(segments.size());
        for (final Collection<F> segment : segments) {
            Callable<F> callable = new Callable<F>() {
                @Override
                public F call() throws Exception {
                    return doReduce(reduceFunction, segment);
                }
            };
            FutureTask<F> futureTask = new FutureTask<F>(callable);
            executionContext.getExecutor().execute(futureTask);
            futureSegmentResults.add(futureTask);
        }

        // merge future results
        Collection<F> results = new ArrayList<F>(segments.size());
        for (Future<F> future : futureSegmentResults) {
            try {
                F segmentResult = future.get();
                results.add(segmentResult);
            } catch (InterruptedException e) {
                // propagate
                Thread.currentThread().interrupt();
                abort(e, futureSegmentResults);
            } catch (ExecutionException e) {
                abort(e.getCause(), futureSegmentResults);
            }
        }

        /* this would be better implemented as a recursive asynchronous function,
         * but as long as the number of segments are small (smaller than the size of each segment)
         * this method will actually be faster */

        return doReduce(reduceFunction, results);
    }

    @Override
    protected <T> ParallelFunctional<T> withData(Collection<T> data) {
        return new ParallelSegmented<T>(data, executionContext);
    }

    @Override
    protected ParallelFunctional<F> withExecutionContext(ExecutionContext executionContext) {
        return new ParallelSegmented<F>(data, executionContext);
    }
}
